package com.spx.sql;

import com.spx.chapter05.pojo.Event;
import com.spx.util.ClickSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.time.Duration;

/**
 * create by undeRdoG on  2022-05-10  15:44
 * 凡心所向，素履以往，生如逆旅，一苇以航。
 */
public class TableAndStreamExchange {

    public static void main(String[] args) throws Exception {
        /*
        *  表和流的转换，就还是要基于 StreamTableEnvironment ，基于单纯的TableEnvironment是不行的
        * */

        StreamExecutionEnvironment StreamEnv = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(StreamEnv);

        SingleOutputStreamOperator<Event> dataStream = StreamEnv.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                );


        Table eventTable = streamTableEnvironment.fromDataStream(dataStream);


        streamTableEnvironment.createTemporaryView("clickTable", eventTable);

        // 聚合打印
        Table result = streamTableEnvironment.sqlQuery("select `user`,count(*) as cnt from clickTable group by `user`");

        //Table sink 'default_catalog.default_database.Unregistered_DataStream_Sink_1' doesn't support consuming update changes which is produced by node GroupAggregate(groupBy=[user], select=[user, COUNT(*) AS cnt])
       // toDataStream 是不支持更新操作的, 可以使用 toChangelogStream
        DataStream<Row> resultStream = streamTableEnvironment.toChangelogStream(result);

        resultStream.print();

        StreamEnv.execute();

    }
}
